package com.bungarus.busi;

import com.bungarus.cluster.ClusterManager;
import com.bungarus.cluster.NodeInfo;
import com.bungarus.codec.protobuf.MessageProtobuf;
import com.bungarus.model.*;
import com.bungarus.offline.OfflineMessagePersist;
import com.bungarus.utils.IPAddressUtils;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.protobuf.ProtobufDecoder;
import io.netty.handler.codec.protobuf.ProtobufEncoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;
import io.netty.handler.timeout.IdleStateHandler;
import org.apache.log4j.Logger;
import org.springframework.data.mongodb.core.MongoTemplate;

import java.util.List;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by tang on 2018/11/17.
 */
public class InnerClient implements BaseClient{
    Logger logger = Logger.getLogger(InnerClient.class);

    protected static final int RECONNECT_INTERVAL = 2;
    protected static final int RETRY_INIT_COUNT = 1;
    protected static final int RETRY_TOTAL_COUNT = 2;

    private static final String RETRY_FAILED_MESSAGE_COLLECTION = "retry-failed-message";
    private final String uuid;

    protected Channel channel;
    protected EventLoopGroup eventLoopGroup;

    protected CopyOnWriteArrayList<NodeInfo> queue;

    private AtomicInteger index;
    private String clusterTag;
    private int clusterId;

    private final MongoTemplate mongoTemplate;

    private final AbstractBungarusServer hostedServer;
    /**
     * The remote server the client connected to
     */
    private final IPAddressUtils.HostPortPair remoteServer = new IPAddressUtils.HostPortPair();

    /**
     * Node id of server the client connected to
     */
    private int remoteNodeId = -1;

    protected ConcurrentHashMap<String, Channel> clientChannels = new ConcurrentHashMap<>();
    protected ConcurrentHashMap<Long, AtomicInteger> retryQueue = new ConcurrentHashMap<>();
    /**
     * cluster tag the connection will be built from
     */
    String connectionFromWhichCluster;
    /**
     * mongo operator for persisting the retry-failed message
     */
    private RetryFailedMessageRepository retryFailedMessageRepository;
    private volatile boolean readyTobeShutdown = false;

    public InnerClient(AbstractBungarusServer hostedServer, int index, CopyOnWriteArrayList<NodeInfo> queue, MongoTemplate mongoTemplate) {
        this.index = new AtomicInteger(index);
        this.queue = queue;
        this.clusterTag = queue.get(index).getClusterBelongsTo();
        this.clusterId = queue.get(index).getClusterIDBelongsTo();
        this.mongoTemplate = mongoTemplate;
        this.retryFailedMessageRepository = new RetryFailedMessageRepository(mongoTemplate);
        this.hostedServer = hostedServer;
        this.uuid = UUID.randomUUID().toString().toLowerCase().replace("-", "");
    }

    @Override
    public Channel getChannel() {
        return channel;
    }

    @Override
    public int getRemoteNodeId() {
        return remoteNodeId;
    }

    public int getClusterId() { return clusterId; }

    public void connect(int startFrom, int endTo, String connectionFrom){
        if(readyTobeShutdown) {
            logger.info("refuse to connect node as waiting for shutting down.");
            return;
        }
        if(null != this.eventLoopGroup) {
            this.eventLoopGroup.shutdownGracefully();
            this.eventLoopGroup = null;
        }
        if(startFrom >= endTo) {
            throw new RuntimeException("too much times to connect node error.");
        }
        final int count = startFrom + 1;
        this.eventLoopGroup = new NioEventLoopGroup();

        this.connectionFromWhichCluster = connectionFrom;
        NodeInfo node = queue.get(this.index.get());

        Bootstrap client = new Bootstrap();
        client.group(this.eventLoopGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
                .handler(new MessageChannelInitializer(this));
        try {
            this.remoteServer.host = node.getIp();
            this.remoteServer.port = node.getMessagePorts().get(this.connectionFromWhichCluster);

            ChannelFuture channelFuture = client.connect(this.remoteServer.host, this.remoteServer.port);
            channelFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (!channelFuture.isSuccess()) {
                        channelFuture.channel().eventLoop().schedule(() ->  {
                            hostedServer.unregisterInnerClientConnectionMap(node.getClusterBelongsTo(), remoteServer, InnerClient.this);
                            connect(count, endTo, connectionFrom);
                            }, RECONNECT_INTERVAL, TimeUnit.SECONDS) ;
                    }
                    else if(channelFuture.isSuccess()){
                        channel = channelFuture.sync().channel();
                        clientChannels.put(node.getClusterBelongsTo(), channel);
                        remoteNodeId = node.getNodeId();
                        hostedServer.registerInnerClientConnectionMap(node.getClusterBelongsTo(), remoteServer, InnerClient.this);
                    }
                }
            });
        }
        catch (Exception e) {
            e.printStackTrace();
            try {
                Thread.sleep(5000);
                connect(count, endTo, connectionFrom);
            }
            catch (InterruptedException e1) {
                e1.printStackTrace();
            }
        }
    }

    @Override
    public void connect(AtomicInteger index, EventLoopGroup eventLoop) throws BungarusClientException{
        if(readyTobeShutdown) {
            logger.info("refuse to connect node as waiting for shutting down.");
            return;
        }
        if(null == queue || queue.size() <= 0){
            throw new BungarusClientException("queue of remote servers is empty.");
        }
        if(null == eventLoop) {
            this.eventLoopGroup = new NioEventLoopGroup();
        }
        else {
            this.eventLoopGroup.shutdownGracefully();
            this.eventLoopGroup = null;
            this.eventLoopGroup = new NioEventLoopGroup();
        }

        index.compareAndSet(-1, queue.size() - 1);

        NodeInfo node = queue.get(index.get());
        Bootstrap client = new Bootstrap();
        client.group(this.eventLoopGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
                .handler(new MessageChannelInitializer(this));
        try {
            this.remoteServer.host = node.getIp();
            this.remoteServer.port = node.getMessagePorts().get(this.connectionFromWhichCluster);

            ChannelFuture channelFuture = client.connect(this.remoteServer.host, this.remoteServer.port);
            channelFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (!channelFuture.isSuccess()) {
                        channelFuture.channel().eventLoop().schedule(() -> {
                            hostedServer.unregisterInnerClientConnectionMap(node.getClusterBelongsTo(), remoteServer, InnerClient.this);
                            int n = InnerClient.this.index.get() - 1;
                            if (n < 0) {
                                //Re-get cluster nodes of the specified cluster
                                InnerClient.this.reLoadRemoteServers(channelFuture.channel().eventLoop());
                            }
                            else {
                                try {
                                    InnerClient.this.connectNextNode(channelFuture.channel().eventLoop());
                                }
                                catch (BungarusClientException e) {
                                    e.printStackTrace();
                                }
                            }
                        }, RECONNECT_INTERVAL, TimeUnit.SECONDS);
                    }
                    else if(channelFuture.isSuccess()){
                        channel = channelFuture.sync().channel();
                        clientChannels.put(node.getClusterBelongsTo(), channel);
                        remoteNodeId = node.getNodeId();
                        hostedServer.registerInnerClientConnectionMap(node.getClusterBelongsTo(), remoteServer, InnerClient.this);
                    }
                }
            });
        }
        catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void run(String connectionFrom) throws BungarusClientException {
        this.connectionFromWhichCluster = connectionFrom;
        connect(index, eventLoopGroup);
    }

    @Override
    public void connectNextNode(EventLoopGroup eventLoopGroup) throws BungarusClientException {
        if(readyTobeShutdown) {
            logger.info("refuse to connect next node as waiting for shutting down.");
            return;
        }
        logger.info("connect the next node which is one of nodes of cluster tagged as " + this.clusterTag);
        if(queue.size() > 0) {
            queue.remove(index.get());
            index.decrementAndGet();
            connect(index, eventLoopGroup);
        }
    }

    @Override
    public void reLoadRemoteServers(EventLoopGroup eventLoopGroup) {
        logger.info("re-connect all nodes of cluster tagged as " + this.clusterTag);
        if(eventLoopGroup.isShutdown()) {
            new Thread(() -> {
                try {
                    reconnectRemoteServers(null);
                }
                catch (Exception e) {
                    try {
                        Thread.sleep(RECONNECT_INTERVAL * 1000);
                        reLoadRemoteServers(eventLoopGroup);
                    }
                    catch (InterruptedException e1) {
                        e1.printStackTrace();
                    }
                }
            }).start();
        }
        else {
            eventLoopGroup.submit(() -> {
                try {
                    reconnectRemoteServers(eventLoopGroup);
                }
                catch (Exception e) {
                    try {
                        Thread.sleep(RECONNECT_INTERVAL * 1000);
                        reLoadRemoteServers(eventLoopGroup);
                    }
                    catch (InterruptedException e1) {
                        e1.printStackTrace();
                    }
                }
            });
        }
    }

    private void reconnectRemoteServers(EventLoopGroup loopGroup) throws Exception {
        ClusterManager clusterManager = new ClusterManager(clusterTag);
        List<NodeInfo> nodes = clusterManager.nodes(clusterId);
        queue.clear();
        queue.addAll(nodes);
        queue = hostedServer.calcRemoteServerNodes(queue);
        index.set(queue.size() - 1);
        connect(index, loopGroup);
    }

    @Override
    public void sendMessage(MessageProtobuf.Message msg, String clusterTag) throws BungarusClientException {
        sendMessage(msg, clusterTag, null);
    }

    @Override
    public void sendMessage(MessageProtobuf.Message msg, String clusterTag, ChannelHandlerContext forAck) throws BungarusClientException {
        if(readyTobeShutdown) {
            throw new BungarusClientException("can not send message as being shutdown.");
        }
        Channel connectionChannel = this.clientChannels.get(clusterTag);
        if(null == connectionChannel || !connectionChannel.isActive()) {
            throw new BungarusClientException("no channel can be used to send message.");
        }
        connectionChannel.writeAndFlush(msg).addListener(channelFuture -> {
            //after sending message, the message is putted into a queue to wait for being acknowledged
            if(channelFuture.isDone() && channelFuture.isSuccess()) {
                if(null != forAck) {
                    forAck.writeAndFlush(MessageBuilder.buildAckMessage(msg.getHeader().getId(), msg.getHeader().getId2()));
                }
                DischargeQueue dischargeQueue = Discharger.getQueueInstance(); // generic queue
                dischargeQueue.addToQueue(msg);
            }
            else {//retry
                Long id = Long.valueOf(msg.getHeader().getId());
                int e;
                if(null == retryQueue.get(id)){
                    AtomicInteger ai = new AtomicInteger(RETRY_INIT_COUNT);
                    e = ai.getAndIncrement();
                    retryQueue.put(id, ai);
                }
                else {
                    e = retryQueue.get(id).getAndIncrement();
                    if(e > RETRY_TOTAL_COUNT) {
                        //save the message into persistent storage (supported by mongo)
                        saveTryFailedMessage(msg);
                        ///////////////////////////////////////////////////////////
                        try {
                            connectionChannel.close();
                            this.clientChannels.remove(clusterTag);
                            this.connectNextNode(this.eventLoopGroup);
                        }
                        catch (BungarusClientException e1){
                            this.reLoadRemoteServers(this.eventLoopGroup);
                        }
                        finally {
                            retryQueue.remove(id);
                        }
                        return;
                    }
                }
                Thread.sleep((long)Math.pow(2, e) * 1000);
                sendMessage(msg, clusterTag, forAck);
            }
        });
    }

    @Override
    public void saveTryFailedMessage(MessageProtobuf.Message failedMsg) {
        RetryFailedMessage.Builder builder = RetryFailedMessage.newBuilder();
        builder.id(Long.parseLong(failedMsg.getHeader().getId()))
                .app(failedMsg.getHeader().getApp())
                .clusterTag(this.clusterTag)
                .content(failedMsg.getBody().getContent())
                .dev(failedMsg.getHeader().getDev())
                .from(failedMsg.getHeader().getFrom())
                .gaction(failedMsg.getHeader().getGaction())
                .lang(failedMsg.getHeader().getLang())
                .priority(failedMsg.getHeader().getPriority())
                .sentTime(failedMsg.getHeader().getSentTime())
                .tenant(failedMsg.getHeader().getTenantID())
                .to(failedMsg.getHeader().getTo())
                .type(failedMsg.getHeader().getType());

        retryFailedMessageRepository.add(builder.built(), RETRY_FAILED_MESSAGE_COLLECTION);
    }

    @Override
    public boolean isShuttingDown() {
        return readyTobeShutdown;
    }

    @Override
    public void close() {
        readyTobeShutdown = true;
        if (null != this.eventLoopGroup) {
            remoteNodeId = -1;
            try {
                if (!this.eventLoopGroup.isTerminated()
                        && null != this.clientChannels.get(this.clusterTag)
                        && this.clientChannels.get(this.clusterTag).isOpen()) {
                    this.clientChannels.get(this.clusterTag).close();
                }
            }
            finally {
                this.eventLoopGroup.shutdownGracefully();
            }
        }
        this.retryQueue.clear();
        this.clientChannels.clear();
        this.hostedServer.unregisterInnerClientConnectionMap(clusterTag, remoteServer, this);
    }

    public void unregisterInnerClientConnectionMap(IPAddressUtils.HostPortPair remoteServer) {
        this.hostedServer.unregisterInnerClientConnectionMap(remoteServer, this);
    }

    public String getSrvIp() {
        return this.remoteServer.host;
    }

    public int getSrvPort() {
        return this.remoteServer.port;
    }

    public CopyOnWriteArrayList<NodeInfo> getQueue() {
        return queue;
    }

    public void addToQueue(int index, NodeInfo node) {
        queue.add(index, node);
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        InnerClient that = (InnerClient) o;
        return uuid.equals(that.uuid);
    }

    @Override
    public int hashCode() {
        return Objects.hash(uuid);
    }

    @Override
    public String toString() {
        return getClass().getCanonicalName() + "@uuid:" + uuid;
    }

    protected class MessageChannelInitializer extends ChannelInitializer<SocketChannel> {
        private static final int ALL_IDLE_TIMEOUT = 10;

        private final InnerClient client;

        public MessageChannelInitializer(InnerClient client) {
            this.client = client;
        }

        @Override
        protected void initChannel(SocketChannel socketChannel) throws Exception {
            socketChannel.pipeline().addLast(new IdleStateHandler(0,
                    0, ALL_IDLE_TIMEOUT, TimeUnit.SECONDS));
            socketChannel.pipeline().addLast(new ProtobufVarint32FrameDecoder());
            socketChannel.pipeline().addLast(new ProtobufDecoder(MessageProtobuf.Message.getDefaultInstance()));
            socketChannel.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());
            socketChannel.pipeline().addLast(new ProtobufEncoder());
            socketChannel.pipeline().addLast(new NodeAsClientMessageHandler((MessageProtobuf.Message offlineMsg) -> {
                OfflineMessagePersist persist = OfflineMessagePersist.getInstance(InnerClient.this.mongoTemplate);
                try {
                    persist.doBusiness(offlineMsg);
                }
                catch (BusiProcessingException e) {
                    e.printStackTrace();
                }
                return null;
            }));
            socketChannel.pipeline().addLast(new ClientHeartbeatHandler(client));
        }
    }
}
